package com.wu.rocketmq_demo.demo;

import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.nio.charset.StandardCharsets;


/**
 * @author benjamin_5
 * @Description 异步发送
 * @date 2024/2/25
 */
public class Producer2AsyncDemo {

    public static void main(String[] args) throws MQClientException, MQBrokerException, RemotingException, InterruptedException {
        DefaultMQProducer producer = new DefaultMQProducer("group_test");

        producer.setNamesrvAddr("127.0.0.1:9876");

        // 启动实例
        producer.start();

        for (int i = 0; i < 10; i++) {
            Message msg = new Message("topic_test", "tag_test", ("hello "+ i).getBytes(StandardCharsets.UTF_8));
            producer.send(msg, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    System.out.println(new String(msg.getBody(), StandardCharsets.UTF_8)+"发送结果："+sendResult);
                }

                @Override
                public void onException(Throwable e) {
                    System.out.println("异常："+e);
                    e.printStackTrace();
                }
            });

        }
        // 等待发送完再结束主线程
        Thread.sleep(10000);

        producer.shutdown();

    }
}
